package controller

import (
	"context"
	"encoding/json"
	"errors"
	v1 "k8s.io/api/apps/v1"
	v13 "k8s.io/api/core/v1"
	v12 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/dynamic/dynamicinformer"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/restmapper"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
	"operator/pkg/types/item"
	"operator/pkg/types/mesh"
	"operator/pkg/types/resource"
	"operator/pkg/types/unstruct"
	"operator/pkg/util"
	rt "runtime"
	"sync"
	"time"
)

type PodController struct {
	client       dynamic.Interface
	ctx          context.Context
	mu           sync.RWMutex
	cfg          *rest.Config
	workerCount  int
	maxRetry     int
	informer     dynamicinformer.DynamicSharedInformerFactory
	queue        workqueue.RateLimitingInterface
	mapper       *restmapper.DeferredDiscoveryRESTMapper
	pLister      cache.GenericLister
	resourceDict *resource.ResourceMap
	updateCh     chan []byte
	addIpCh      chan string
}

func (m *PodController) Run(stopCh chan struct{}) {
	for i := 0; i < m.workerCount; i++ {
		go wait.Until(m.worker, time.Minute, stopCh)
	}
}

func (m *PodController) List() []byte {
	return []byte(m.resourceDict.Resource())
}

func (m *PodController) worker() {
	for m.processNextItem() {
	}
}

func (m *PodController) syncPod(it item.ItemOperation) error {
	//fmt.Println("sync pod name:" + it.Item)
	switch it.Operation {
	case item.ADD:
		obj, err := m.pLister.Get(it.Item)
		if err != nil {
			return err
		}

		pod := &v13.Pod{}
		err = unstruct.UnstructToObject(obj.(*unstructured.Unstructured), pod)
		if err != nil {
			return err
		}

		uid := string(pod.GetUID())
		resoucesStr, ok := pod.GetAnnotations()[util.ResourceAnnotation]
		if !ok {
			m.addIpCh <- pod.Status.PodIP
			return nil
		}
		resources := make([]*mesh.Resource, 0, 10)
		err = json.Unmarshal([]byte(resoucesStr), &resources)
		if err != nil {
			return err
		}
		similar, err := m.resourceDict.Add(uid, &resource.PodResource{
			Resource: resources,
			IP:       pod.Status.PodIP,
		})
		if err != nil {
			return err
		}
		if similar {
			return nil
		}
		m.updateCh <- []byte{}
		m.addIpCh <- pod.Status.PodIP
	case item.DEL:
		has, err := m.resourceDict.Del(it.Item)
		if err != nil {
			return err
		}
		if !has {
			return nil
		}
		m.updateCh <- []byte{}
	}

	return nil
}

func (m *PodController) addPod(obj interface{}) {
	switch obj.(type) {
	case *v13.Pod:
		if _, ok := obj.(*v13.Pod).Annotations[util.ResourceAnnotation]; ok && obj.(*v13.Pod).Status.Phase == v13.PodRunning {
			ms, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
			if err != nil {
				return
			}
			op := item.ItemOperation{
				Operation: item.ADD,
				Item:      ms,
			}
			m.queue.Add(op)
		} else if _, ok := obj.(*v13.Pod).Labels[util.ResourceAnnotation]; ok && obj.(*v13.Pod).Status.Phase == v13.PodRunning {
			ms, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
			if err != nil {
				return
			}
			op := item.ItemOperation{
				Operation: item.ADD,
				Item:      ms,
			}
			m.queue.Add(op)
		}
	case *unstructured.Unstructured:
		pod := &v13.Pod{}
		err := unstruct.UnstructToObject(obj.(*unstructured.Unstructured), pod)
		if err != nil {
			return
		}
		if _, ok := pod.Annotations[util.ResourceAnnotation]; ok && pod.Status.Phase == v13.PodRunning {
			ms, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
			if err != nil {
				return
			}
			op := item.ItemOperation{
				Operation: item.ADD,
				Item:      ms,
			}
			m.queue.Add(op)
		} else if _, ok := pod.Labels[util.ResourceAnnotation]; ok && pod.Status.Phase == v13.PodRunning {
			ms, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
			if err != nil {
				return
			}
			op := item.ItemOperation{
				Operation: item.ADD,
				Item:      ms,
			}
			m.queue.Add(op)
		}
	}
}

func (m *PodController) delPod(obj interface{}) {
	switch obj.(type) {
	case *v13.Pod:
		op := item.ItemOperation{
			Operation: item.DEL,
			Item:      string(obj.(*v13.Pod).GetUID()),
		}
		m.queue.Add(op)
	case *unstructured.Unstructured:
		pod := &v13.Pod{}
		err := unstruct.UnstructToObject(obj.(*unstructured.Unstructured), pod)
		if err != nil {
			return
		}
		op := item.ItemOperation{
			Operation: item.DEL,
			Item:      string(pod.GetUID()),
		}
		m.queue.Add(op)
	}

}

func (m *PodController) processNextItem() bool {
	it, shutdown := m.queue.Get()
	if shutdown {
		return false
	}
	defer m.queue.Done(it)

	err := m.syncPod(it.(item.ItemOperation))
	if err != nil {
		return false
	}
	return true
}

func (m *PodController) getGVR(gvk schema.GroupVersionKind) *schema.GroupVersionResource {
	restmapper, err := m.mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
	if err != nil {
		return nil
	}

	return &restmapper.Resource
}

func (m *PodController) getGVRFromObject(obj runtime.Object) *schema.GroupVersionResource {
	gvk := schema.GroupVersionKind{}
	switch obj.(type) {
	case *mesh.Mesh:
		gvk.Version = "v1"
		gvk.Group = "king.io"
		gvk.Kind = "mesh"
	case *v13.Pod:
		gvk.Version = "v1"
		gvk.Kind = "pod"
	case *v1.Deployment:
		gvk.Version = "v1"
		gvk.Group = "apps"
		gvk.Kind = "deployment"
	case *v13.Service:
		gvk.Version = "v1"
		gvk.Kind = "service"
	}

	restmapper, err := m.mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
	if err != nil {
		return nil
	}

	return &restmapper.Resource
}

func (c *PodController) GetMySelfIp() (string, error) {
	if rt.GOOS == "windows" {
		return "192.168.1.99", nil
	}

	now := 0
	for {
		now++
		if now == 6 {
			return "", errors.New("get ip error")
		}
		gvr := c.getGVRFromObject(&v13.Pod{})
		data, err := c.client.Resource(*gvr).Namespace("mesh-ns").List(c.ctx, v12.ListOptions{
			LabelSelector: "app=mesh",
		})
		if err != nil {
			return "", err
		}

		pod := &v13.Pod{}
		err = unstruct.UnstructToObject(&data.Items[0], pod)
		if err != nil {
			return "", err
		}
		if pod.Status.PodIP == "" {
			time.Sleep(time.Second * 10)
			now++
			continue

		}

		return pod.Status.PodIP, nil
	}

}

func (c *PodController) GetTraceIp() (string, error) {
	//if rt.GOOS == "windows" {
	//	return "192.168.1.99", nil
	//}

	now := 0
	for {
		now++
		if now == 6 {
			return "", errors.New("get ip error")
		}
		gvr := c.getGVRFromObject(&v13.Service{})
		data, err := c.client.Resource(*gvr).Namespace("").List(c.ctx, v12.ListOptions{
			LabelSelector: "app.kubernetes.io/component=jaeger-agent",
		})
		if err != nil {
			return "", err
		}

		svc := &v13.Service{}
		err = unstruct.UnstructToObject(&data.Items[0], svc)
		if err != nil {
			return "", err
		}
		if svc.Spec.ClusterIP == "" {
			time.Sleep(time.Second * 10)
			now++
			continue
		}

		return svc.Spec.ClusterIP, nil
	}

}

func NewPodController(ctx context.Context,
	config *rest.Config,
	dSharedInformer dynamicinformer.DynamicSharedInformerFactory,
	mapper *restmapper.DeferredDiscoveryRESTMapper,
	client dynamic.Interface,
	workerCount int,
	maxRetry int,
	updateCh chan []byte,
	addIp chan string,
) *PodController {
	c := &PodController{
		ctx:          ctx,
		cfg:          config,
		client:       client,
		maxRetry:     maxRetry,
		workerCount:  workerCount,
		informer:     dSharedInformer,
		mapper:       mapper,
		queue:        workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "podController"),
		mu:           sync.RWMutex{},
		resourceDict: resource.NewResourceMap(),
		updateCh:     updateCh,
		addIpCh:      addIp,
	}

	dSharedInformer.ForResource(*c.getGVRFromObject(&v13.Pod{})).Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: c.addPod,
		UpdateFunc: func(oldObj, newObj interface{}) {
			c.addPod(newObj)
		},
		DeleteFunc: func(obj interface{}) {
			c.delPod(obj)
		},
	})

	c.pLister = dSharedInformer.ForResource(*c.getGVRFromObject(&v13.Pod{})).Lister()

	return c
}
